package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
)

func CreateTopic(addr string, tp ...string) error {
	broker := sarama.NewBroker(addr)
	err := broker.Open(nil)
	if err != nil {
		return err
	}
	defer func() {
		err = broker.Close()
		if err != nil {
			fmt.Println(err)
		}
	}()
	tr := &sarama.CreateTopicsRequest{}
	tr.TopicDetails = make(map[string]*sarama.TopicDetail)
	for _, s := range tp {
		tr.TopicDetails[s] = &sarama.TopicDetail{NumPartitions: 1, ReplicationFactor: 1}
	}
	resp, err := broker.CreateTopics(tr)
	if err != nil {
		return err
	}
	for _, err = range resp.TopicErrors {
		if err != nil {
			return err
		}
	}
	return nil
}

func TopicList(addr string) ([]string, error) {
	client, err := sarama.NewClient([]string{addr}, nil)
	if err != nil {
		return nil, err
	}
	return client.Topics()
}

func DeleteTopic(addr string, tp ...string) error {
	broker := sarama.NewBroker(addr)
	err := broker.Open(nil)
	if err != nil {
		return err
	}
	defer func() {
		err = broker.Close()
		if err != nil {
			fmt.Println(err)
		}
	}()
	tr := &sarama.DeleteTopicsRequest{}
	tr.Topics = tp
	resp, err := broker.DeleteTopics(tr)
	if err != nil {
		return err
	}
	for _, err = range resp.TopicErrorCodes {
		if err != nil {
			return err
		}
	}
	return nil
}
